package cn.tedu.work;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

//模拟工作队列模式中的消费者
public class WorkConsumer {
    public static void main(String[] args) throws IOException, TimeoutException {
        //1.创建连接工厂对象
        ConnectionFactory cf = new ConnectionFactory();
        cf.setHost("192.168.8.100");
        cf.setPort(5672);   //默认端口号，可以省略不写
        cf.setUsername("test");
        cf.setPassword("test");
        cf.setVirtualHost("/");  //如果不指定虚拟主机，默认是"/",可以省略不写
        //2.通过连接工厂对象创建tcp连接对象，通过tcp连接对象创建channel
        Connection nc = cf.newConnection();
        Channel cc = nc.createChannel();
        //3.通过channel创建队列
        cc.queueDeclare("work.queue", true, false, false, null);
        //4.消费数据
        //4.1编写消息处理的回调函数
        DeliverCallback deliverCallback = new DeliverCallback() {
            @Override
            public void handle(String s, Delivery delivery) throws IOException {
                //delivery里面封装了要处理的消息
                byte[] body = delivery.getBody();
                //转成对应数据类型的数据进行处理
                String msg = new String(body);
                System.out.println(s + "收到了数据：" + msg);
                System.out.println(s + "数据开始处理");
                for (int i = 0; i < msg.length(); i++) {
                    //模拟处理繁忙的场景，每遇到一个"."，就sleep一秒
                    if (msg.charAt(i) == '.') {
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
                System.out.println(s + "处理完毕");
                //消费者要主动向服务器端发送消息回执
                long ackTag = delivery.getEnvelope().getDeliveryTag();
                //第二个参数：是否发送全部消息的回执消息
                cc.basicAck(ackTag, false);
            }
        };
        //4.2编写取消接收数据的回调函数
        CancelCallback cancelCallback = new CancelCallback() {
            @Override
            public void handle(String s) throws IOException {
            }
        };
        //实现合理分发，只修改第二个参数autoACK的值为false，无法实现手动ack
        //还需要设置，消费者同时只能接受一条数据
        cc.basicQos(1);
        cc.basicConsume("work.queue", false, deliverCallback, cancelCallback);
    }
}
